// AnomalyDetectionMapper.java
package com.taxi.analysis.anomalyDetection;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 高价行程异常检测Mapper
 * 功能：识别和分类异常高价行程
 * 技术：条件过滤 + 自定义分组
 */
public class AnomalyDetectionMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    // 异常检测阈值配置
    private static final double HIGH_FARE_THRESHOLD = 100.0;  // 高价阈值
    private static final double EXTREME_FARE_THRESHOLD = 200.0; // 极高价阈值
    private static final double HIGH_FARE_PER_MILE_THRESHOLD = 10.0; // 单位距离高价阈值

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过CSV标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 解析CSV数据行
            String[] fields = parseCSVLine(line);

            // CSV字段索引（根据enhanced_data_cleaning.py的输出格式）:
            // 0:tpep_pickup_datetime, 1:tpep_dropoff_datetime, 2:PULocationID, 3:DOLocationID,
            // 4:pickup_zone, 5:dropoff_zone, 6:trip_distance, 7:total_amount, 8:passenger_count,
            // 9:fare_per_mile, 10:price_group, 11:is_anomaly, 12:is_iqr_anomaly, 13:is_zscore_anomaly

            if (fields.length >= 14) {
                // 提取关键字段
                String pickupZone = fields[4].trim();
                String dropoffZone = fields[5].trim();
                double tripDistance = parseDouble(fields[6]);
                double totalAmount = parseDouble(fields[7]);
                int passengerCount = parseInt(fields[8]);
                double farePerMile = parseDouble(fields[9]);
                String priceGroup = fields[10].trim();
                boolean isAnomaly = parseBoolean(fields[11]);
                boolean isIQRAnomaly = parseBoolean(fields[12]);
                boolean isZScoreAnomaly = parseBoolean(fields[13]);

                // 条件过滤：只处理有效的行程数据
                if (isValidTrip(pickupZone, dropoffZone, tripDistance, totalAmount)) {

                    // 自定义分组：根据异常类型和严重程度分组
                    String anomalyType = classifyAnomalyType(totalAmount, farePerMile,
                            isAnomaly, isIQRAnomaly, isZScoreAnomaly);

                    // 创建复合键：异常类型_地区组合
                    String compositeKey = anomalyType + "_" + getRegionGroup(pickupZone);

                    // 创建详细的异常信息
                    String anomalyInfo = createAnomalyInfo(pickupZone, dropoffZone,
                            tripDistance, totalAmount,
                            passengerCount, farePerMile,
                            priceGroup, anomalyType);

                    outputKey.set(compositeKey);
                    outputValue.set(anomalyInfo);
                    context.write(outputKey, outputValue);
                }
            }

        } catch (Exception e) {
            // 记录错误但继续处理其他行
            System.err.println("处理行时出错: " + value.toString());
            System.err.println("错误信息: " + e.getMessage());
        }
    }

    /**
     * 验证行程数据的有效性
     */
    private boolean isValidTrip(String pickupZone, String dropoffZone,
                                double tripDistance, double totalAmount) {
        return !pickupZone.isEmpty() &&
                !dropoffZone.isEmpty() &&
                !pickupZone.equals("null") &&
                !dropoffZone.equals("null") &&
                tripDistance > 0 &&
                totalAmount > 0 &&
                tripDistance <= 200 && // 合理的行程距离上限
                totalAmount <= 1000;   // 合理的费用上限
    }

    /**
     * 异常类型分类器
     * 使用条件过滤技术对异常进行多维度分类
     */
    private String classifyAnomalyType(double totalAmount, double farePerMile,
                                       boolean isAnomaly, boolean isIQRAnomaly,
                                       boolean isZScoreAnomaly) {

        // 多层条件判断，实现精细化异常分类
        if (totalAmount >= EXTREME_FARE_THRESHOLD) {
            if (farePerMile >= HIGH_FARE_PER_MILE_THRESHOLD) {
                return "极高价_高单价"; // 既总价极高又单价极高
            } else {
                return "极高价_长距离"; // 总价极高但可能是长距离导致
            }
        } else if (totalAmount >= HIGH_FARE_THRESHOLD) {
            if (farePerMile >= HIGH_FARE_PER_MILE_THRESHOLD) {
                return "高价_高单价"; // 高价且单价异常
            } else {
                return "高价_正常单价"; // 高价但单价正常
            }
        } else if (farePerMile >= HIGH_FARE_PER_MILE_THRESHOLD) {
            return "正常价_高单价"; // 总价正常但单价异常（可能短距离高收费）
        } else if (isZScoreAnomaly && isIQRAnomaly) {
            return "统计异常_双重"; // 两种统计方法都认为异常
        } else if (isZScoreAnomaly) {
            return "统计异常_Z分数"; // Z-Score方法检测到的异常
        } else if (isIQRAnomaly) {
            return "统计异常_IQR"; // IQR方法检测到的异常
        } else if (isAnomaly) {
            return "综合异常"; // 其他类型的异常
        } else {
            return "正常行程"; // 正常行程作为对照组
        }
    }

    /**
     * 地区分组器
     * 将具体地区归类到更大的区域组
     */
    private String getRegionGroup(String zone) {
        if (zone.startsWith("Manhattan")) {
            return "Manhattan";
        } else if (zone.startsWith("Brooklyn")) {
            return "Brooklyn";
        } else if (zone.startsWith("Queens")) {
            return "Queens";
        } else if (zone.startsWith("Bronx")) {
            return "Bronx";
        } else if (zone.startsWith("Staten_Island")) {
            return "Staten_Island";
        } else if (zone.contains("Airport")) {
            return "Airport";
        } else {
            return "Other";
        }
    }

    /**
     * 创建详细的异常信息字符串
     */
    private String createAnomalyInfo(String pickupZone, String dropoffZone,
                                     double tripDistance, double totalAmount,
                                     int passengerCount, double farePerMile,
                                     String priceGroup, String anomalyType) {

        return String.format("%s|%s|%.2f|%.2f|%d|%.2f|%s|%s",
                pickupZone, dropoffZone, tripDistance, totalAmount,
                passengerCount, farePerMile, priceGroup, anomalyType);
    }

    /**
     * 解析CSV行，处理可能的引号和逗号
     */
    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    /**
     * 处理包含引号的CSV行
     */
    private String[] parseQuotedCSV(String line) {
        java.util.List<String> fields = new java.util.ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    /**
     * 安全的boolean解析
     */
    private boolean parseBoolean(String str) {
        try {
            return Boolean.parseBoolean(str.trim()) || "True".equalsIgnoreCase(str.trim());
        } catch (Exception e) {
            return false;
        }
    }
}
